package lagou.test;

import com.google.gson.Gson;
import lagou.mapper.ProductsMapper;
import lagou.resultmapper.ESSearchResultMapper;
import org.apache.http.HttpHost;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.query.FetchSourceFilter;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.test.context.junit4.SpringRunner;
import lagou.pojo.Products;

import java.io.IOException;
import java.util.*;

/**
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class test {

    @Autowired
    private ProductsMapper productsMapper;


    //定义数据
    public long insert(Products products){
        products.setName("哈哈");
        products.setPrice(200.00);
        products.setFlag("22");
        products.setGoodsDese("1");
        products.setImages("ss");
        products.setGoodsStock(1);
        products.setGoodsType("1");
        productsMapper.insert(products);
        return products.getId();
    }

    //生产数据
    @Test
    public void product(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.164.20:9092,192.168.164.20:9093,192.168.164.20:9094");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaProducer<String, String> producer = new KafkaProducer<String,String>(properties);

        //定义主题
        String topic = "product";
        //获得消息的内容
        String msg = String.valueOf(insert(new Products()));

        //构建一个消息对象：指定主题和消息
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic,msg);

        //发送
        producer.send(producerRecord);

        System.out.println("消息发送成功，传入的商品id是："+msg);


        producer.close();

    }

    //消费数据（消息）
    @Test
    public void consumer() throws IOException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.164.20:9092,192.168.164.20:9093,192.168.164.20:9094");

        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //指定分组的名称
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"lagou_group1");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        //定义主题
        String topic = "product";
        //订阅消息
        consumer.subscribe(Collections.singletonList(topic));


        while(true){

            ConsumerRecords<String, String> poll = consumer.poll(500);//500就是超时时间
            for(ConsumerRecord<String,String> consumer1 : poll){

                System.out.println("主题："+consumer1.topic());
                System.out.println("偏移量："+consumer1.offset());
                System.out.println("消息："+consumer1.value());
                Products products = productsMapper.selectById(consumer1.value());
                System.out.println(products);
                ines(products);
            }
            System.out.println("========");




        }

    }

    // Json工具
    private Gson gson = new Gson();

    private RestHighLevelClient client;

    //添加文档
    public void ines(Products products) throws IOException {
        String source = gson.toJson(products);
        IndexRequest request =
                new IndexRequest("lagou","product",String.valueOf(products.getId()));
        request.source(source, XContentType.JSON);
        IndexResponse response = client.index(request);
        System.out.println(response);
    }
    @Before
    public void init() {


        RestClientBuilder restClientBuilder = RestClient.builder(
                new HttpHost("192.168.164.20", 9200, "http")

        );
        client = new RestHighLevelClient(restClientBuilder);


    }

    @After
    public void close() throws IOException {
        // 关闭客户端

        client.close();
    }




    @Test
    //查询文档
    public void sele() throws IOException {
        // 创建搜索对象
        SearchRequest request = new SearchRequest();
        // 查询构建工具
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchQuery("name", "哈哈"));
        sourceBuilder.sort("price", SortOrder.ASC);

        sourceBuilder.from(0);
        sourceBuilder.size(5);

        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("name");//高亮的字段
        highlightBuilder.requireFieldMatch(false);//是否多个字段都高亮
        highlightBuilder.preTags("<span style='color:red'>");//前缀后缀
        highlightBuilder.postTags("</span>");
        sourceBuilder.highlighter(highlightBuilder);

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.rangeQuery("price").gt(0).lt(0));

        sourceBuilder.postFilter(boolQueryBuilder);
        //传递语句并执行
        basicQuery(sourceBuilder);
    }
    /*
      定义公用方法，来操作添加语句并执行
       */
    private void basicQuery(SearchSourceBuilder sourceBuilder) throws IOException {

        // 创建搜索对象
        SearchRequest request = new SearchRequest();

        //添加语句
        request.source(sourceBuilder);

        // 搜索
        SearchResponse response = client.search(request);
        System.out.println(response);


        SearchHits hits = response.getHits();
        SearchHit[] searchHits = hits.getHits();
        for (SearchHit hit : searchHits) {
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            for (Map.Entry<String, HighlightField> highlightField : highlightFields.entrySet()) {
                // 获取高亮的Key
                String key = highlightField.getKey();
                // 获取高亮的Value，这就是我们操作添加的高亮信息
                HighlightField value = highlightField.getValue();
                Text[] fragments = value.getFragments();
                sourceAsMap.put(key, fragments[0].toString());
            }

            Products item = gson.fromJson(gson.toJson(sourceAsMap), Products.class);
            System.out.println("item = " + item);
        }
    }







}
